package com.qf.config;

import com.qf.mapper.MessageMapper;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 作者：大狗
 * 时间：2021/7/12 20:57
 */
@Configuration
public class MQInit {
    //队列名称
    private static final String QUEUE_NAME = "pay_queue";
    //交换机名称
    public static final String PAY_EXCHANGE = "pay_exchange";
    //routingKey
    public static final String PAY_KEY = "pay_key";
    //这个集合可以处理并发问题
    private LinkedBlockingQueue<String> lbq = new LinkedBlockingQueue<>();
    //注入连接工厂对象
    @Autowired
    private ConnectionFactory factory;
    @Autowired
    private MessageMapper messageMapper;

    //spring的RabbitTemplate对象中没有提供回调方法，需要自己定义
    @Bean("availableTemplate")
    public RabbitTemplate rabbitTemplate(){
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(factory);
        //开启消息没有到达队列的回调 yml已经配置
        //rabbitTemplate.setMandatory(true);

        //消息是否到达交换机的回调方法
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                //获取数据id
                String id = correlationData.getId();
                //成功到达交换机，并不代表到达MQ中
                if(b){
                    if (lbq.contains(id)) {
                        lbq.remove(id); //删除id，防止内存溢出
                        messageMapper.updateTryCount(id);
                        //SQL操作try_count次数+1

                    }else {
                        // 将本地投递的消息删除掉,删除到数据表中的记录
                        messageMapper.deleteById(id);
                    }
                }else {
                    // 如果没有正确到达MQ的消息, 我们不理它, 我们将其交给消息补偿系统来做.
                    //SQL操作try_count次数+1
                    messageMapper.updateTryCount(id);
                }
            }
        });
        //消息到达队列前的方法回调
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            // 该方法是消息没有到达队列的时候回调, 如果没有交换机, 该方法不会回调.
            @Override
            public void returnedMessage(Message message, int i, String s, String s1, String s2) {
                // 消息没有到达队列的时候, 将 CorrelationId 添加到 LinkedList 中
                lbq.add(message.getMessageProperties().getCorrelationId());
            }
        });
        return rabbitTemplate;
    }

}
